package com.darrenchan.spark.rdd

import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper
import org.joda.time.DateTime

import scala.collection.JavaConversions._
import scala.collection.mutable.ListBuffer
import scala.util.matching.Regex

object SparkStreamingBaiduTest {
    def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("WordCountApp")
        val sc = new SparkContext(sparkConf)

        //业务逻辑
        //    val counts = sc.textFile("hello.txt").
        //      flatMap(_.split(" ")).
        //      map((_, 1)).
        //      reduceByKey(_+_)

        val clusterRegex: Regex = """cluster:\w+""".r
        val pipeletRegex: Regex = """piplet: \d+""".r
        val startPointRegex: Regex = """offset: \d+""".r
        val commandNoRegex = """\"command_no\":\d+""".r
        val logTimeReg: Regex = """\"tm\":\"\d{13}\"""".r

        val rdd = sc.textFile("streaming.txt").flatMap(row => {
            //      val cluster = clusterRegex.findFirstIn(row) match {
            //        case Some(y) => y.substring(8)
            //        case _ => ""
            //      }
            //      val pipelet = pipeletRegex.findFirstIn(row) match {
            //        case Some(y) => y.substring(8).toInt
            //        case _ => 0
            //      }
            //      val startPoint = startPointRegex.findFirstIn(row) match {
            //        case Some(y) => y.substring(8).toLong
            //        case _ => 0l
            //      }
            //      val commandNo = commandNoRegex.findFirstIn(row) match {
            //        case Some(y) => y.substring(13)
            //        case _ => ""
            //      }
            //      val tm = logTimeReg.findFirstIn(row) match {
            //        case Some(y) =>
            //          y.substring(6, 19).toLong
            //        case _ => 0.toLong
            //      }
            //      val logTime = if (tm > 0) {
            //        new DateTime(tm).toString("yyyyMMdd")
            //      } else {
            //        ""
            //      }

            val dataStr = row.split("""data":\[""")(1).split("""\],"ubc_id""")(0)
            val dataNode = new ObjectMapper().readTree(s"[${dataStr}]")

            var nidJsonList: List[String] = Nil

            if (row.contains("\"ubc_id\":\"10023\"")) {
                val nidListMulti: List[String] = dataNode.toList.map(item => {
                    try {
                        val content = item.get("content").toString
                        val id = item.get("id").asText()
                        val contentNode = new ObjectMapper().readTree(content)
                        if (contentNode.has("ext")) {
                            if (contentNode.get("ext").has("nid")) {
                                val nidStr = contentNode.get("ext").get("nid").toString
                                val nidNode = new ObjectMapper().readTree(nidStr)
//                                val size = nidNode.size()
//                                val nids: StringBuffer = new StringBuffer()
//                                for (i <- 0 until size - 1) {
//                                    nids.append(nidNode.get(i).get("rid").asText())
//                                    nids.append(",")
//                                }
//                                nids.append(nidNode.get(size - 1).get("rid").asText())

                                val nidList = nidNode.map(_.get("rid").asText()).toList
                                (nidList.mkString(","), id)
                            } else {
                                ("", id)
                            }
                        } else {
                            ("", "")
                        }
                    } catch {
                        case e: Exception => {
                            ("", "")
                        }
                    }
                }).filter(item => item._1.contains("dt") && item._1.length() > 0 && item._2.equals("10023")).map(item => {
                    item._1
                })

//                val nidList: ListBuffer[String] = new ListBuffer[String]
//                for (i <- 0 until nidListMulti.size) {
//                    val nids = nidListMulti.get(i).split(",")
//                    for (j <- 0 until nids.size) {
//                        nidList.append(nids(j))
//                    }
//                }
//                nidJsonList = nidList.toList
                nidJsonList = nidListMulti.flatMap(_.split(","))
                println(nidJsonList.size)
            }

            nidJsonList.map(nid => nid)
            //(cluster, pipelet, startPoint, commandNo, logTime, tm, episodeId)
            //idsJsonList.map(vid => (cluster, pipelet, startPoint, commandNo, logTime, tm, vid))
        })

        println(rdd.collect().mkString("\n"))

        sc.stop()
    }
}
